package com.cold.demo.rabbitmq.demo2.$6.rpc;

import com.cold.demo.rabbitmq.utils.ConnectionUtils;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.UUID;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeoutException;

public class RPCClient implements AutoCloseable {

    private Connection connection;
    private Channel    channel;
    private String     requestQueueName = "rpc_queue";


    public RPCClient() throws IOException, TimeoutException {
        connection = ConnectionUtils.getConnection();
        channel = connection.createChannel();
    }

    public static void main(String[] argv) {
        try (RPCClient client = new RPCClient()) {
            for (int i = 0; i < 32; i++) {
                Thread.sleep(300);
                client.call("消息--：" + i);
            }
        } catch (IOException | TimeoutException | InterruptedException e) {
            e.printStackTrace();
        }
    }

    public String call(String message) throws IOException, InterruptedException {
        final String corrId = UUID.randomUUID().toString();

        // 当我们不向 queueDeclare() 提供任何参数时，我们将创建一个非持久的、独占的、自动删除的队列，并使用生成的名称：
        // 这个是用于RPCServer响应客户端的时候  将消息发送至该队列
        String replyQueueName = channel.queueDeclare().getQueue();

        AMQP.BasicProperties props = new AMQP.BasicProperties
                .Builder()
                .correlationId(corrId)
                .replyTo(replyQueueName)
                .build();

        channel.basicPublish("", requestQueueName, props, message.getBytes(StandardCharsets.UTF_8));

        final BlockingQueue<String> response = new ArrayBlockingQueue<>(1);

        // 监听响应队列
        String ctag = channel.basicConsume(replyQueueName, true, (consumerTag, delivery) -> {
            if (delivery.getProperties().getCorrelationId().equals(corrId)) {
                response.offer(new String(delivery.getBody(), StandardCharsets.UTF_8));
            }
        }, consumerTag -> {
        });

        // 获取响应数据，如果一直没有获取到就一直堵塞在这里
        System.out.println("获取返回数据");
        String result = response.take();
        System.out.println("获取返回数据成功：" + result);
        channel.basicCancel(ctag);
        return result;
    }

    public void close() throws IOException {
        connection.close();
    }
}